/*
 * (c) Copyright 2014 Hewlett-Packard Development Company, L.P. All rights reserved.
 */
package com.hp.snap.gr.db.voltdb.sp;

import com.hp.snap.gr.db.voltdb.sp.VoltDBTable.VoltDBRow;
import com.hp.snap.gr.exception.DatabaseException;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * <code>@Statistics</code> Returns statistics about the usage of the VoltDB database.
 * <p/>
 * <b>Syntax</b>
 * <p/>
 * <code>@Statistics</code> String component, Integer delta-flag
 * <p/>
 * <b>Description</b>
 * <p/>
 * The <code>@Statistics</code> system procedure returns information about the VoltDB database. The second argument,
 * component, specifies what aspect of VoltDB to return statistics about. The third argument, delta-flag, specifies
 * whether statistics are reported from when the database started or since the last call to <code>@Statistics</code>
 * where the flag was set.
 * <p/>
 * If the delta-flag is set to zero, the system procedure returns statistics since the database started. If the
 * deltaflag is non-zero, the system procedure returns statistics for the interval since the last time
 * <code>@Statistics</code> was called with a non-zero flag. (If <code>@Statistics</code> has not been called with a
 * non-zero flag before, the first call with the flag set returns statistics since startup.)
 * <p/>
 * Note that in a cluster with K-safety, if a node fails, the statistics reported by this procedure are reset to zero
 * for the node when it rejoins the cluster.
 * <p/>
 * The following are the allowable values of component:
 * <p/>
 * <b>"DR"</b> Returns information about the status of database replication, including how much data is waiting to be
 * sent to the DR agent. This information is available only if the database is licensed for database replication.
 * <p/>
 * <b>Return Values</b>
 * <p/>
 * Returns different VoltTables depending on which component is requested. The following tables identify the structure
 * of the return values for each component. (Note that the MANAGEMENT component returns seven VoltTables.)
 * <p/>
 * <b>DR</b> - Returns two VoltTables. The first table contains information about the replication streams, which consist
 * of a row per partition for each server. The data shows the current state of replication and how much data is
 * currently queued for the DR agent.
 * <p/>
 * <pre>
 * <b>TIMESTAMP</b> BIGINT The timestamp when the information was collected (in milliseconds).
 *
 * <b>HOST_ID</b> INTEGER Numeric ID for the host node.
 *
 * <b>HOSTNAME</b> STRING Server name of the host node.
 *
 * <b>PARTITION_ID</b> INTEGER The numeric ID for the logical partition.
 *
 * <b>STREAMTYPE</b> STRING The type of stream, which can either be "TRANSACTIONS" or "SNAPSHOT".
 *
 * <b>TOTALBYTES</b> BIGINT The total number of bytes currently queued for transmission to the DR agent.
 *
 * <b>TOTALBYTESINMEMORY</b> BIGINT The total number of bytes of queued data currently held in memory.
 * If the amount of total bytes is larger than the amount in memory, the remainder is kept in overflow storage on disk.
 *
 * <b>TOTALBUFFERS</b> BIGINT The total number of buffers in this partition currently waiting for acknowledgement from the DR agent.
 * Partitions create a buffer every five milliseconds.
 *
 * <b>LASTACKTIMESTAMP</b> BIGINT The timestamp of the last acknowledgement received from the DR agent.
 *
 * <b>ISSYNCED</b> STRING A text string indicating whether the database is currently being replicated.
 * If replication has not started, or the overflow capacity has been exceeded (that is, replication has failed),
 * the value of ISSYNCED is "false". If replication is currently in progress, the value is "true".
 *
 * <b>MODE</b> STRING A text string indicating whether this particular partition is replicating data for the DR agent ("NORMAL") or not ("PAUSED").
 * Only one copy of each logical partition actually sends data to the DR agent during replication.
 * So for clusters with a K-safety value greater than zero, not all physical partitions will report "NORMAL" even when replication is in progress.
 * </pre>
 * <p/>
 * The second table returns a row for every host in the cluster, showing whether a replication snapshot is in progress
 * and if it is, the status of transmission to the DR agent.
 * <p/>
 * <pre>
 * <b>TIMESTAMP</b> BIGINT The timestamp when the information was collected (in milliseconds).
 *
 * <b>HOST_ID</b> INTEGER Numeric ID for the host node.
 *
 * <b>HOSTNAME</b> STRING Server name of the host node.
 *
 * <b>ENABLED</b> STRING A text string indicating whether the database is currently being replicated. Possible values are "true" and "false".
 *
 * <b>SYNCSNAPSHOTSTATE</b> STRING A text string indicating the current state of the synchronization snapshot that begins replication.
 * During normal operation, this value is "NOT_SYNCING" indicating either that replication is not active or that transactions are actively being replicated.
 * If a synchronization snapshot is in progress, this value provides additional infomation about the specific activity underway.
 *
 * <b>ROWSINSYNCSNAPSHOT</b> BIGINT Reserved for future use.
 *
 * <b>ROWSACKEDFORSYNCSNAPSHOT</b> BIGINT Reserved for future use.
 * </pre>
 *
 * @author Yang, Lin
 * @since 3.2
 */
public class StatisticsDRResponses {
    private static final Logger logger = LoggerFactory.getLogger(StatisticsDRResponses.class);

    public static JSONObject getReplicationStreams(JSONObject response) throws DatabaseException {
        if (logger.isTraceEnabled()) {
            logger.trace("VoltDB Table[0]: Replication Streams");
            logger.trace("The response:{}", response.toString());
        }
        try {
            JSONArray results = response.getJSONArray("results");
            if (results.length() < 1) {
                return null;
            }
            JSONObject voltdbTable = results.getJSONObject(0);
            if (logger.isTraceEnabled()) {
                traceVoltDBTable(voltdbTable);
            }
            return voltdbTable;
        } catch (JSONException e) {
            throw new DatabaseException("Decode VoltDB Table (Replication Streams) failed!", e);
        }
    }

    public static VoltDBTable getReplicatingStreams(JSONObject streams) throws DatabaseException {
        //VoltDBTable result = new VoltDBTable("ReplicatingSnapshot");
        VoltDBTable result = new VoltDBTable("ReplicatingStreams");
        try {
            JSONArray schema = streams.getJSONArray("schema");
            List<String> schemaNames = new ArrayList<String>(schema.length());
            for (int ii = 0, nn = schema.length(); ii < nn; ii++) {
                JSONObject schemaRow = schema.getJSONObject(ii);
                String name = schemaRow.getString("name");
                int type = schemaRow.getInt("type");
                schemaNames.add(schemaRow.getString("name"));
                result.putHeader(name, type);
            }

            JSONArray data = streams.getJSONArray("data");
            for (int ii = 0, nn = data.length(); ii < nn; ii++) {
                JSONArray dataRow = data.getJSONArray(ii);
                int _ISSYNCED = schemaNames.indexOf("ISSYNCED");
                int _MODE = schemaNames.indexOf("MODE");
                boolean ISSYNCED = dataRow.getBoolean(_ISSYNCED);
                String MODE = dataRow.getString(_MODE);
                if (ISSYNCED && ReplicationMode.NORMAL.name().equals(MODE)) {
                    VoltDBRow row = new VoltDBRow();
                    for (int jj = 0, mm = dataRow.length(); jj < mm; jj++) {
                        String name = schemaNames.get(jj);
                        String value = String.valueOf(dataRow.get(jj));
                        row.put(name, value);
                    }
                    result.addRow(row);
                }
            }
        } catch (JSONException e) {
            throw new DatabaseException("Decode replication streams failed!", e);
        }
        return result;
    }

    public static int getSyncNotFinishedCount(JSONObject streams) throws DatabaseException {

        VoltDBTable result = new VoltDBTable("ReplicatingStreams");
        try {
            JSONArray schema = streams.getJSONArray("schema");
            List<String> schemaNames = new ArrayList<String>(schema.length());
            for (int ii = 0, nn = schema.length(); ii < nn; ii++) {
                JSONObject schemaRow = schema.getJSONObject(ii);
                String name = schemaRow.getString("name");
                int type = schemaRow.getInt("type");
                schemaNames.add(schemaRow.getString("name"));
                result.putHeader(name, type);
            }

            JSONArray data = streams.getJSONArray("data");
            for (int ii = 0, nn = data.length(); ii < nn; ii++) {
                JSONArray dataRow = data.getJSONArray(ii);
                int _ISSYNCED = schemaNames.indexOf("ISSYNCED");
                int _MODE = schemaNames.indexOf("MODE");
                boolean ISSYNCED = dataRow.getBoolean(_ISSYNCED);
                String MODE = dataRow.getString(_MODE);
                if (!ISSYNCED && ReplicationMode.NORMAL.name().equals(MODE)) {
                    VoltDBRow row = new VoltDBRow();
                    for (int jj = 0, mm = dataRow.length(); jj < mm; jj++) {
                        String name = schemaNames.get(jj);
                        String value = String.valueOf(dataRow.get(jj));
                        row.put(name, value);
                    }
                    result.addRow(row);
                }
            }
        } catch (JSONException e) {
            throw new DatabaseException("Decode replication streams failed!", e);
        }
        return result.getRows().size();

    }

    public static JSONObject getReplicationSnapshot(JSONObject response) throws DatabaseException {
        if (logger.isTraceEnabled()) {
            logger.trace("VoltDB Table: Replication Snapshot");
        }
        try {
            JSONArray results = response.getJSONArray("results");
            if (results.length() < 2) {
                return null;
            }
            JSONObject voltdbTable = results.getJSONObject(1);
            if (logger.isTraceEnabled()) {
                traceVoltDBTable(voltdbTable);
            }
            return voltdbTable;
        } catch (JSONException e) {
            throw new DatabaseException("Decode VoltDB Table (Replication Snapshot) failed!", e);
        }
    }

    public static VoltDBTable getReplicatingSnapshot(JSONObject snapshot) throws DatabaseException {
        //VoltDBTable result = new VoltDBTable("ReplicatingStreams");
        VoltDBTable result = new VoltDBTable("ReplicatingSnapshot");
        try {
            JSONArray schema = snapshot.getJSONArray("schema");
            List<String> schemaNames = new ArrayList<String>(schema.length());
            for (int ii = 0, nn = schema.length(); ii < nn; ii++) {
                JSONObject schemaRow = schema.getJSONObject(ii);
                String name = schemaRow.getString("name");
                int type = schemaRow.getInt("type");
                schemaNames.add(name);
                result.putHeader(name, type);
            }

            JSONArray data = snapshot.getJSONArray("data");
            for (int ii = 0, nn = data.length(); ii < nn; ii++) {
                JSONArray dataRow = data.getJSONArray(ii);
                int _ENABLED = schemaNames.indexOf("ENABLED");
                boolean ENABLED = dataRow.getBoolean(_ENABLED);
                int _SYNCSNAPSHOTSTATE = schemaNames.indexOf("SYNCSNAPSHOTSTATE");
                String SYNCSNAPSHOTSTATE = dataRow.getString(_SYNCSNAPSHOTSTATE);
                if (ENABLED && "NOT_SYNCING".equals(SYNCSNAPSHOTSTATE)) {
                    VoltDBRow row = new VoltDBRow();
                    for (int jj = 0, mm = dataRow.length(); jj < mm; jj++) {
                        String name = schemaNames.get(jj);
                        String value = String.valueOf(dataRow.get(jj));
                        row.put(name, value);
                    }
                    result.addRow(row);
                }
            }
        } catch (JSONException e) {
            throw new DatabaseException("Decode replication streams failed!", e);
        }
        return result;
    }

    public static void traceVoltDBTable(JSONObject voltdbTable) throws DatabaseException {
        try {
            JSONArray schema = voltdbTable.getJSONArray("schema");
            List<String> schemaNames = new ArrayList<String>(schema.length());
            for (int ii = 0, nn = schema.length(); ii < nn; ii++) {
                JSONObject schemaRow = schema.getJSONObject(ii);
                schemaNames.add(schemaRow.getString("name"));
            }

            JSONArray data = voltdbTable.getJSONArray("data");
            for (int ii = 0, nn = data.length(); ii < nn; ii++) {
                JSONArray dataRow = data.getJSONArray(ii);
                StringBuilder sb = new StringBuilder();
                sb.append("Row[").append(ii).append("]: ");
                for (int jj = 0, mm = dataRow.length(); jj < mm; jj++) {
                    String column = String.valueOf(dataRow.get(jj));
                    String name = schemaNames.get(jj);
                    sb.append(name).append("=").append(column).append(", ");
                }
                if (dataRow.length() > 0) {
                    sb.delete(sb.length() - 2, sb.length() - 1);
                }
                logger.trace(sb.toString());
            }
        } catch (JSONException e) {
            throw new DatabaseException("Decode VoltDB Table failed!", e);
        }
    }

    public enum ReplicationMode {
        PAUSED,
        NORMAL
    }
}
